package dse.messages.multiversion;

import java.util.Hashtable;

import dse.DataObject;
import dse.Experiment;
import dse.ExperimentRecorder;
import dse.Node;
import dse.messages.Message;
import dse.messages.MessageHandler;
import dse.messages.MessagePurpose;
import dse.messages.MessageSender;
import dse.messages.MessageType;
import dse.Version;

public class ReadCompleteTLOMessageHandler extends MessageSender implements
		MessageHandler {
	private DataObject dobj;
	private VersionReader reader;
	private boolean global;
	
	public ReadCompleteTLOMessageHandler(Experiment experiment, DataObject dobj, boolean g, VersionReader reader)
	{
		super(experiment);
		
		this.dobj = dobj;
		this.reader = reader;
		this.global = g;
	}

	@Override
	public boolean acceptMessage(Message message) {
		// TODO Auto-generated method stub
		return true;
	}

	@Override
	public void messageInterupted(ExperimentRecorder recorder, Message message,
			double time) {
		// TODO Auto-generated method stub
		//recorder.debug(this, "msgIntrpt: " + message, time);
		//reader.readVersion(recorder, null, time, message.to());
	}

	@Override
	public void messageReceived(ExperimentRecorder recorder, Message message,
			double time) {
		// TODO Auto-generated method stub
		recorder.debug(this, "msgRcvd: " + message, time);
		if (message.messageType() != MessageType.ReadCompleteTLO) {
			recorder.warn(
					"ReadCompleteTLOMessageHandler recieved " + message.messageType() + " message."
					);
			return;
		}			
		// What do we know about the share?
		Node vs = message.to();
		
		// Does the receiver have the share?
		Version bv = vs.removeCurReads(dobj, reader.getTransaction().id());
		if(bv != null)
		{
			Hashtable<Node, Integer> residentNodes = null;
			if(global)
				residentNodes = experiment.modules().assigner().hostedShares(experiment.network(), dobj);
			else
				residentNodes = experiment.modules().assigner().hostedShares(experiment.network().lookupCluster(vs.clusterId()), dobj);
			
			for(Node node : residentNodes.keySet())
			{
				Message m = super.createSVRMessage(vs, node, dobj, 
						bv, MessagePurpose.DataObjectUpdate);
				recorder.debug(this, "new msg: " + m, time);
				experiment.messageQueue().enqueueMessage(m, time);
			}
		}
	}
}
